package com.sptus.mqtt;

import java.text.SimpleDateFormat;
import java.util.ArrayList;
import java.util.Date;
import java.util.List;
import java.util.concurrent.TimeUnit;

import javax.annotation.PostConstruct;

import com.sptus.service.IEmailService;
import org.eclipse.paho.client.mqttv3.IMqttDeliveryToken;
import org.eclipse.paho.client.mqttv3.MqttCallback;
import org.eclipse.paho.client.mqttv3.MqttClient;
import org.eclipse.paho.client.mqttv3.MqttConnectOptions;
import org.eclipse.paho.client.mqttv3.MqttException;
import org.eclipse.paho.client.mqttv3.MqttMessage;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
import org.springframework.scheduling.annotation.Async;

import com.alibaba.fastjson.JSONObject;
import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
import com.sptus.common.constant.CommonConstant;
import com.sptus.common.core.redis.RedisCache;
import com.sptus.framework.mqtt.MyMqttClient;
import com.sptus.service.IotIEquipmentService;
import com.sptus.system.domain.BBWS;
import com.sptus.system.domain.DevInfoMsg0;
import com.sptus.system.domain.DevInfoMsg0History;
import com.sptus.system.domain.DevInfoMsg1;
import com.sptus.system.domain.DevInfoMsg1History;
import com.sptus.system.domain.DevInfoMsg2;
import com.sptus.system.domain.DevInfoMsg2History;
import com.sptus.system.domain.DevInfoMsg3;
import com.sptus.system.domain.DevInfoMsg3History;
import com.sptus.system.domain.DevInfoMsg4;
import com.sptus.system.domain.DevInfoMsg4History;
import com.sptus.system.domain.DevInfoMsg5;
import com.sptus.system.domain.DevInfoMsg5History;
import com.sptus.system.domain.DevInfoMsg6;
import com.sptus.system.domain.Equipment;
import com.sptus.system.domain.alarm.AlarmParser;
import com.sptus.system.service.IConfigInfoService;
import com.sptus.system.service.IDevInfoMsg0HistoryService;
import com.sptus.system.service.IDevInfoMsg0Service;
import com.sptus.system.service.IDevInfoMsg1HistoryService;
import com.sptus.system.service.IDevInfoMsg1Service;
import com.sptus.system.service.IDevInfoMsg2HistoryService;
import com.sptus.system.service.IDevInfoMsg2Service;
import com.sptus.system.service.IDevInfoMsg3HistoryService;
import com.sptus.system.service.IDevInfoMsg3Service;
import com.sptus.system.service.IDevInfoMsg4HistoryService;
import com.sptus.system.service.IDevInfoMsg4Service;
import com.sptus.system.service.IDevInfoMsg5HistoryService;
import com.sptus.system.service.IDevInfoMsg5Service;
import com.sptus.system.service.IDevInfoMsg6Service;
import com.sptus.system.service.ISptDevLogService;
import com.sptus.system.service.impl.WorkOrdersServiceImpl;

import cn.hutool.core.util.ObjectUtil;
import lombok.SneakyThrows;
import lombok.extern.slf4j.Slf4j;

//import static com.sptus.mqtt.IotRefreshJob.REDIS_DURING_KEY_FORMAT;
//import static com.sptus.mqtt.IotRefreshJob.REDIS_SN_LIST_KEY;

/**
 * Mqtt连接回调函数 ，用来接收已经订阅的消息
 *
 * @author Mr.Liu
 * @date 2021/4/1 9:36
 */
@Component
@Slf4j
//@Profile({"dev","local"})
public class MqttIotCallback implements MqttCallback {


    @Autowired
    IDevInfoMsg0Service devInfoMsg0Service;
    @Autowired
    IConfigInfoService configInfoService;
    @Autowired
    IDevInfoMsg1Service devInfoMsg1Service;
    @Autowired
    IDevInfoMsg2Service devInfoMsg2Service;
    @Autowired
    IDevInfoMsg3Service devInfoMsg3Service;
    @Autowired
    IDevInfoMsg4Service devInfoMsg4Service;
    @Autowired
    IDevInfoMsg5Service devInfoMsg5Service;
    @Autowired
    IDevInfoMsg6Service devInfoMsg6Service;

    @Autowired
    IDevInfoMsg0HistoryService devInfoMsg0HistoryService;
    @Autowired
    IDevInfoMsg2HistoryService devInfoMsg2HistoryService;
    @Autowired
    IDevInfoMsg1HistoryService devInfoMsg1HistoryService;
    @Autowired
    IDevInfoMsg3HistoryService devInfoMsg3HistoryService;
    @Autowired
    IDevInfoMsg4HistoryService devInfoMsg4HistoryService;
    @Autowired
    IDevInfoMsg5HistoryService devInfoMsg5HistoryService;
    @Autowired
    ISptDevLogService sptDevLogService;
    @Autowired
    WorkOrdersServiceImpl workOrdersService;
    @Autowired
    IotIEquipmentService equipmentService;
    @Autowired
    RedisCache redisUtil;
    @Autowired
    IEmailService emailService;

    public static MqttIotCallback mqttReceiveCallback;

    @SneakyThrows
    @Override
    public void connectionLost(Throwable throwable) {
        log.warn("MQTT连接丢失，开始重连机制, 异常: {}", throwable.getMessage());

        MqttClient client = MyMqttClient.mqttClient;
        MqttConnectOptions option = MyMqttClient.mqttConnectOptions;

        if (client == null || option == null) {
            log.error("MQTT客户端或连接选项为空，无法重连");
            return;
        }

        // 设置重连参数
        option.setCleanSession(false); // 保持会话状态
        option.setAutomaticReconnect(true); // 启用自动重连
        option.setMaxReconnectDelay(30000); // 最大重连延迟30秒
        option.setConnectionTimeout(30); // 连接超时30秒
        option.setKeepAliveInterval(60); // 心跳间隔60秒

        int maxRetryCount = 50;
        int retryCount = 0;
        long retryDelay = 1000; // 初始重连延迟1秒

        while (retryCount < maxRetryCount && !client.isConnected()) {
            try {
                log.info("尝试重连MQTT服务器，第{}次重试...", retryCount + 1);

                // 检查客户端状态
                if (client.isConnected()) {
                    log.info("MQTT客户端已连接，重连成功");
                    break;
                }

                // 尝试重连
                client.connect(option);

                if (client.isConnected()) {
                    log.info("MQTT重连成功，重新订阅主题");
                    try {
                        client.subscribe("/#", 0);
                        log.info("主题重新订阅成功");
                        break;
                    } catch (MqttException e) {
                        log.error("重新订阅主题失败: {}", e.getMessage(), e);
                    }
                }

            } catch (MqttException e) {
                retryCount++;
                log.error("MQTT重连失败，第{}次重试: {}", retryCount, e.getMessage());

                if (retryCount >= maxRetryCount) {
                    log.error("MQTT重连失败，已达到最大重试次数({})，停止重连", maxRetryCount);
                    break;
                }

                // 指数退避策略
                retryDelay = Math.min(retryDelay * 2, 30000);
                log.info("等待{}毫秒后重试...", retryDelay);
                Thread.sleep(retryDelay);
            }
        }

        if (!client.isConnected()) {
            log.error("MQTT重连失败，客户端状态异常");
            // 可以在这里添加告警通知逻辑
            try {
                List<String> targetEmailList = new ArrayList<>();
                targetEmailList.add("314886745@qq.com");
                targetEmailList.add("jzy08707003@126.com");
                emailService.sendDeviceWarningInfoEmail(targetEmailList, "job", "0000", "未更新超时", "MQTT重连失败");
            } catch (Exception e) {
                log.error("发送告警邮件失败: {}", e.getMessage());
            }
        }
    }

    @PostConstruct //通过@PostConstruct实现初始化bean之前进行的操作
    public void init() {
        log.info("初始化");
        mqttReceiveCallback = this;
        mqttReceiveCallback.devInfoMsg0Service = this.devInfoMsg0Service;
        mqttReceiveCallback.devInfoMsg1Service = this.devInfoMsg1Service;
        mqttReceiveCallback.devInfoMsg2Service = this.devInfoMsg2Service;
        mqttReceiveCallback.devInfoMsg3Service = this.devInfoMsg3Service;
        mqttReceiveCallback.devInfoMsg4Service = this.devInfoMsg4Service;
        mqttReceiveCallback.devInfoMsg5Service = this.devInfoMsg5Service;
        mqttReceiveCallback.devInfoMsg6Service = this.devInfoMsg6Service;
        mqttReceiveCallback.configInfoService = this.configInfoService;
        mqttReceiveCallback.devInfoMsg2HistoryService = this.devInfoMsg2HistoryService;
        mqttReceiveCallback.devInfoMsg3HistoryService = this.devInfoMsg3HistoryService;
        mqttReceiveCallback.devInfoMsg1HistoryService = this.devInfoMsg1HistoryService;
        mqttReceiveCallback.devInfoMsg0HistoryService = this.devInfoMsg0HistoryService;
        mqttReceiveCallback.devInfoMsg4HistoryService = this.devInfoMsg4HistoryService;
        mqttReceiveCallback.devInfoMsg5HistoryService = this.devInfoMsg5HistoryService;
        mqttReceiveCallback.workOrdersService = this.workOrdersService;
        mqttReceiveCallback.sptDevLogService = this.sptDevLogService;
        mqttReceiveCallback.redisUtil = this.redisUtil;
        mqttReceiveCallback.emailService = this.emailService;
    }

    /**
     * 发布消息
     *
     * @param publishTopic 发布消息的主题名称
     * @param message      消息内容
     */
    public static void publishMessage(String publishTopic, String message) {
        log.info("发布消息.......publishTopic:{} message:{}", publishTopic, message);
//            try {
//            	String[] split = publishTopic.split("/");
//                String key = split.length > 1 ? split[1] : "";
//                if(StringUtils.isEmpty(key)) {
//                	 KafkaProducerService.sendMessage(com.sptus.mqtt.KafkaIotCallback.TOPIC_KAFKA, message);
//                }else {
//                	 KafkaProducerService.sendMessage(com.sptus.mqtt.KafkaIotCallback.TOPIC_KAFKA, key, message);
//                }
//                log.info("✅ 已发送到 Kafka：" + message);
//            } catch (Exception e) {
//            	log.error("❌ Kafka 发送失败：" + e.getMessage());
//            }
    }

    @Override
    public void messageArrived(String topic, MqttMessage message) throws Exception {
        //String s = new String(message.getPayload());
        //publishMessage(topic, s);
        messageArrivedOld(topic, message);
    }

    //@Override
    public void messageArrivedOld(String topic, MqttMessage message) throws Exception {
        // 生成13位无特殊符号的唯一标识符
        String processId = String.valueOf(System.currentTimeMillis()).substring(0, 13);
        log.info("messageArrivedOld 开始处理消息, processId: {}, topic: {}, message: {}", processId, topic, message);
        try {
            String s = new String(message.getPayload());

            JSONObject jsonObject = JSONObject.parseObject(new String(message.getPayload()));
            //设备号
            String sn = jsonObject.get("SN").toString();
            //msgType
            String msg = jsonObject.getString("MSG");
//            String snMsgKey = String.format(REDIS_DURING_KEY_FORMAT, sn);
//            Object snMsgValue = mqttReceiveCallback.redisUtil.getCacheObject(snMsgKey);
//
//            long currentTime = System.currentTimeMillis();
//            boolean shouldSendEmail = false;
//
//            if (ObjectUtil.isEmpty(snMsgValue)) {
//                //将sn加入redis ListKey
//                addSnToRedisList(mqttReceiveCallback.redisUtil,sn);
//                // Redis 键不存在，说明是第一次收到该设备该类型的消息
//                log.info("设备首次收到消息, processId: {}, SN: {}, MSG: {}", processId, sn, msg);
//                // 第一次时也设置 Redis 键，记录当前时间
//                mqttReceiveCallback.redisUtil.setCacheObject(snMsgKey, String.valueOf(currentTime));
//                log.info("首次设置 Redis 键成功, processId: {}, SN: {}, MSG: {}, key: {}", processId, sn, msg, snMsgKey);
//            } else {
//                // 如果不是第一次，则更新 Redis 中的时间戳（永久存储）
//                mqttReceiveCallback.redisUtil.setCacheObject(snMsgKey, String.valueOf(currentTime));
//                log.info("更新 Redis 键时间戳成功, processId: {}, SN: {}, MSG: {}, key: {}", processId, sn, msg, snMsgKey);
//            }

            //判的是为更新数据
            if (jsonObject.get("MSG") == null) {

            } else {

                //设备上线
                LambdaQueryWrapper<Equipment> equipmentLambdaQueryWrapper = new LambdaQueryWrapper<>();
                equipmentLambdaQueryWrapper.eq(Equipment::getSn, jsonObject.get("SN"));
                try {
                    Object o = mqttReceiveCallback.redisUtil.getCacheObject(CommonConstant.SN_KEY + jsonObject.get("SN"));
                    Equipment equipment1 = JSONObject.parseObject(s, Equipment.class);
                    if (ObjectUtil.isEmpty(o)) {
                        List<Equipment> equipments = mqttReceiveCallback.equipmentService.getBaseMapper().selectList(equipmentLambdaQueryWrapper);
                        Equipment equipment = new Equipment();
                        if (equipments.size() > 0) {
                            String[] split = topic.split("/");
                            String s1 = split[1];
                            equipment.setMac(s1);
                            equipment.setPid(equipment1.getPid());
                            equipment.setMn(equipment1.getMn());
                            equipment.setSid(equipment1.getSid());
                            equipment.setStatus("1");
                            mqttReceiveCallback.equipmentService.update(equipment, equipmentLambdaQueryWrapper);
                        } else {
                            String[] split = topic.split("/");
                            String s1 = split[1];
                            equipment.setMac(s1);
                            equipment.setPid(equipment1.getPid());
                            equipment.setMn(equipment1.getMn());
                            equipment.setSid(equipment1.getSid());
                            equipment.setStatus("1");
                            equipment.setSn(sn);
                            equipment.setFailureState("1");
                            equipment.setParallelState("1");
                            equipment.setWorkingCondition(1);
                            mqttReceiveCallback.equipmentService.save(equipment);

                            DevInfoMsg4 devInfoMsg4 = new DevInfoMsg4();
                            devInfoMsg4.setSn(sn);
                            LambdaQueryWrapper<DevInfoMsg4> devInfoMsg4LambdaQueryWrapper = new LambdaQueryWrapper<>();
                            devInfoMsg4LambdaQueryWrapper.eq(DevInfoMsg4::getSn, devInfoMsg4.getSn());
                            DevInfoMsg4 devInfoMsg04 = mqttReceiveCallback.devInfoMsg4Service.getBaseMapper().selectOne(devInfoMsg4LambdaQueryWrapper);
                            if (devInfoMsg04 != null) {
                                mqttReceiveCallback.devInfoMsg4Service.update(devInfoMsg4, devInfoMsg4LambdaQueryWrapper);
                            } else {
                                mqttReceiveCallback.devInfoMsg4Service.save(devInfoMsg4);
                            }
                        }
//                    mqttReceiveCallback.redisUtil.set(CommonConstant.SN_KEY + jsonObject.get("SN"), "",30);
                        mqttReceiveCallback.redisUtil.setCacheObject(CommonConstant.SN_KEY + jsonObject.get("SN"), jsonObject.get("SN"), 6000, TimeUnit.SECONDS);
                    } else {
                        Object o1 = mqttReceiveCallback.redisUtil.getCacheObject(CommonConstant.SN_KEY_UPDATE + jsonObject.get("SN"));
                        Equipment equipment = new Equipment();
                        if (ObjectUtil.isEmpty(o1)) {
                            String[] split = topic.split("/");
                            String s1 = split[1];
                            equipment.setMac(s1);
                            equipment.setPid(equipment1.getPid());
                            equipment.setMn(equipment1.getMn());
                            equipment.setSid(equipment1.getSid());
                            equipment.setStatus("1");
                            mqttReceiveCallback.equipmentService.update(equipment, equipmentLambdaQueryWrapper);
//                        mqttReceiveCallback.redisUtil.set(CommonConstant.SN_KEY_UPDATE + jsonObject.get("SN"),jsonObject.get("SN"),20);
                            mqttReceiveCallback.redisUtil.setCacheObject(CommonConstant.SN_KEY_UPDATE + jsonObject.get("SN"), jsonObject.get("SN"), 120, TimeUnit.SECONDS);
                        }
//                    mqttReceiveCallback.redisUtil.set(CommonConstant.SN_KEY + jsonObject.get("SN"), "",30);
                        mqttReceiveCallback.redisUtil.setCacheObject(CommonConstant.SN_KEY + jsonObject.get("SN"), jsonObject.get("SN"), 6000, TimeUnit.SECONDS);
                    }
                }catch (Exception e){
                    log.info("设备上线异常: {}", e.getMessage());
                }

                log.info("设备数据添加, processId: {}, topic: {}, message: {}", processId, topic, message);
                if (msg.equals("0")) {
                    DevInfoMsg0 devInfoMsg0 = JSONObject.parseObject(new String(s), DevInfoMsg0.class);
                    AlarmParser alarmParser = new AlarmParser();
                    LambdaQueryWrapper<DevInfoMsg0> devInfoMsg0LambdaQueryWrapper = new LambdaQueryWrapper<>();
                    devInfoMsg0LambdaQueryWrapper.eq(DevInfoMsg0::getSn, devInfoMsg0.getSn());
                    DevInfoMsg0 devInfoMsg00 = mqttReceiveCallback.devInfoMsg0Service.getBaseMapper().selectOne(devInfoMsg0LambdaQueryWrapper);
                    DevInfoMsg0History devInfoMsg0History = JSONObject.parseObject(new String(s), DevInfoMsg0History.class);
                    List<Equipment> equipments = mqttReceiveCallback.equipmentService.getBaseMapper().selectList(equipmentLambdaQueryWrapper);

                    // 异步处理 DevInfoMsg0
                    saveDevInfoMsg0Async(processId, devInfoMsg0, devInfoMsg00, devInfoMsg0LambdaQueryWrapper, devInfoMsg0History, equipments, alarmParser);

                } else if (msg.equals("1")) {
                    DevInfoMsg1 devInfoMsg1 = JSONObject.parseObject(new String(s), DevInfoMsg1.class);
                    DevInfoMsg1History devInfoMsg1History = JSONObject.parseObject(new String(s), DevInfoMsg1History.class);
                    LambdaQueryWrapper<DevInfoMsg1> devInfoMsg1LambdaQueryWrapper = new LambdaQueryWrapper<>();
                    devInfoMsg1LambdaQueryWrapper.eq(DevInfoMsg1::getSn, devInfoMsg1.getSn());
                    DevInfoMsg1 devInfoMsg01 = mqttReceiveCallback.devInfoMsg1Service.getBaseMapper().selectOne(devInfoMsg1LambdaQueryWrapper);

                    // 异步处理 DevInfoMsg1
                    saveDevInfoMsg1Async(processId, devInfoMsg1, devInfoMsg01, devInfoMsg1LambdaQueryWrapper, devInfoMsg1History);

                } else if (msg.equals("2")) {
                    DevInfoMsg2 devInfoMsg2 = JSONObject.parseObject(new String(s), DevInfoMsg2.class);
                    DevInfoMsg2History devInfoMsg2History = JSONObject.parseObject(new String(s), DevInfoMsg2History.class);
                    LambdaQueryWrapper<DevInfoMsg2> devInfoMsg2LambdaQueryWrapper = new LambdaQueryWrapper<>();
                    devInfoMsg2LambdaQueryWrapper.eq(DevInfoMsg2::getSn, devInfoMsg2.getSn());
                    DevInfoMsg2 devInfoMsg02 = mqttReceiveCallback.devInfoMsg2Service.getBaseMapper().selectOne(devInfoMsg2LambdaQueryWrapper);

                    // 异步处理 DevInfoMsg2
                    saveDevInfoMsg2Async(processId, devInfoMsg2, devInfoMsg02, devInfoMsg2LambdaQueryWrapper, devInfoMsg2History);

                } else if (msg.equals("3")) {
                    DevInfoMsg3 devInfoMsg3 = JSONObject.parseObject(new String(s), DevInfoMsg3.class);
                    DevInfoMsg3History devInfoMsg3History = JSONObject.parseObject(new String(s), DevInfoMsg3History.class);
                    LambdaQueryWrapper<DevInfoMsg3> devInfoMsg3LambdaQueryWrapper = new LambdaQueryWrapper<>();
                    devInfoMsg3LambdaQueryWrapper.eq(DevInfoMsg3::getSn, devInfoMsg3.getSn());
                    DevInfoMsg3 devInfoMsg03 = mqttReceiveCallback.devInfoMsg3Service.getBaseMapper().selectOne(devInfoMsg3LambdaQueryWrapper);

                    // 异步处理 DevInfoMsg3
                    saveDevInfoMsg3Async(processId, devInfoMsg3, devInfoMsg03, devInfoMsg3LambdaQueryWrapper, devInfoMsg3History);

                } else if (msg.equals("4")) {
                    DevInfoMsg4 devInfoMsg4 = JSONObject.parseObject(new String(s), DevInfoMsg4.class);
                    LambdaQueryWrapper<DevInfoMsg4> devInfoMsg4LambdaQueryWrapper = new LambdaQueryWrapper<>();
                    DevInfoMsg4History devInfoMsg4History = JSONObject.parseObject(new String(s), DevInfoMsg4History.class);
                    devInfoMsg4LambdaQueryWrapper.eq(DevInfoMsg4::getSn, devInfoMsg4.getSn());
                    DevInfoMsg4 devInfoMsg04 = mqttReceiveCallback.devInfoMsg4Service.getBaseMapper().selectOne(devInfoMsg4LambdaQueryWrapper);

                    // 异步处理 DevInfoMsg4
                    saveDevInfoMsg4Async(processId, devInfoMsg4, devInfoMsg04, devInfoMsg4LambdaQueryWrapper, devInfoMsg4History);

                } else if (msg.equals("5")) {
                    DevInfoMsg5 devInfoMsg5 = JSONObject.parseObject(new String(s), DevInfoMsg5.class);
                    DevInfoMsg5History devInfoMsg5History = JSONObject.parseObject(new String(s), DevInfoMsg5History.class);
                    LambdaQueryWrapper<DevInfoMsg5> devInfoMsg5LambdaQueryWrapper = new LambdaQueryWrapper<>();
                    devInfoMsg5LambdaQueryWrapper.eq(DevInfoMsg5::getSn, devInfoMsg5.getSn());
                    DevInfoMsg5 devInfoMsg05 = mqttReceiveCallback.devInfoMsg5Service.getBaseMapper().selectOne(devInfoMsg5LambdaQueryWrapper);

                    // 异步处理 DevInfoMsg5
                    saveDevInfoMsg5Async(processId, devInfoMsg5, devInfoMsg05, devInfoMsg5LambdaQueryWrapper, devInfoMsg5History);

                } else if (msg.equals("6")) {
                    DevInfoMsg6 devInfoMsg6 = JSONObject.parseObject(new String(s), DevInfoMsg6.class);
                    LambdaQueryWrapper<DevInfoMsg6> devInfoMsg6LambdaQueryWrapper = new LambdaQueryWrapper<>();
                    devInfoMsg6LambdaQueryWrapper.eq(DevInfoMsg6::getSn, devInfoMsg6.getSn());
                    DevInfoMsg6 devInfoMsg06 = mqttReceiveCallback.devInfoMsg6Service.getBaseMapper().selectOne(devInfoMsg6LambdaQueryWrapper);

                    // 异步处理 DevInfoMsg6
                    saveDevInfoMsg6Async(processId, devInfoMsg6, devInfoMsg06, devInfoMsg6LambdaQueryWrapper);

                } else if (msg.equals("7")) {
                    DevInfoMsg6 devInfoMsg6 = JSONObject.parseObject(new String(s), DevInfoMsg6.class);
                    LambdaQueryWrapper<DevInfoMsg6> devInfoMsg6LambdaQueryWrapper = new LambdaQueryWrapper<>();
                    devInfoMsg6LambdaQueryWrapper.eq(DevInfoMsg6::getSn, devInfoMsg6.getSn());
                    DevInfoMsg6 devInfoMsg06 = mqttReceiveCallback.devInfoMsg6Service.getBaseMapper().selectOne(devInfoMsg6LambdaQueryWrapper);

                    // 异步处理 DevInfoMsg6
                    saveDevInfoMsg6Async(processId, devInfoMsg6, devInfoMsg06, devInfoMsg6LambdaQueryWrapper);

                } else if (msg.equals("8")) {
                    DevInfoMsg6 devInfoMsg6 = JSONObject.parseObject(new String(s), DevInfoMsg6.class);
                    LambdaQueryWrapper<DevInfoMsg6> devInfoMsg6LambdaQueryWrapper = new LambdaQueryWrapper<>();
                    devInfoMsg6LambdaQueryWrapper.eq(DevInfoMsg6::getSn, devInfoMsg6.getSn());
                    DevInfoMsg6 devInfoMsg06 = mqttReceiveCallback.devInfoMsg6Service.getBaseMapper().selectOne(devInfoMsg6LambdaQueryWrapper);

                    // 异步处理 DevInfoMsg6
                    saveDevInfoMsg6Async(processId, devInfoMsg6, devInfoMsg06, devInfoMsg6LambdaQueryWrapper);

                } else if (msg.equals("10")) {
                    DevInfoMsg6 devInfoMsg6 = JSONObject.parseObject(new String(s), DevInfoMsg6.class);
                    LambdaQueryWrapper<DevInfoMsg6> devInfoMsg6LambdaQueryWrapper = new LambdaQueryWrapper<>();
                    devInfoMsg6LambdaQueryWrapper.eq(DevInfoMsg6::getSn, devInfoMsg6.getSn());
                    DevInfoMsg6 devInfoMsg06 = mqttReceiveCallback.devInfoMsg6Service.getBaseMapper().selectOne(devInfoMsg6LambdaQueryWrapper);

                    // 异步处理 DevInfoMsg6
                    saveDevInfoMsg6Async(processId, devInfoMsg6, devInfoMsg06, devInfoMsg6LambdaQueryWrapper);
                }
            }
//

        } catch (Exception e) {
            log.error("messageArrivedOld 处理消息时发生异常: processId={}, topic={}, error={}", processId, topic, e.getMessage(), e);
            System.out.println(e.fillInStackTrace());
        }
    }

    public static void main(String[] args) {
        System.out.println(new Date().toString());
    }

    @Override
    public void deliveryComplete(IMqttDeliveryToken iMqttDeliveryToken) {

    }

    /**
     * 异步处理 DevInfoMsg0 数据保存
     */
    @Async("mqttAsyncExecutor")
    public void saveDevInfoMsg0Async(String processId, DevInfoMsg0 devInfoMsg0, DevInfoMsg0 devInfoMsg00,
                                     LambdaQueryWrapper<DevInfoMsg0> devInfoMsg0LambdaQueryWrapper,
                                     DevInfoMsg0History devInfoMsg0History,
                                     List<Equipment> equipments,
                                     AlarmParser alarmParser) {
        try {
            log.info("开始异步处理 DevInfoMsg0 数据保存, processId: {}, SN: {}", processId, devInfoMsg0.getSn());

            if (devInfoMsg00 != null) {
                log.info("更新 DevInfoMsg0 数据, processId: {}, data: {}", processId, devInfoMsg0.toString());
                mqttReceiveCallback.devInfoMsg0Service.update(devInfoMsg0, devInfoMsg0LambdaQueryWrapper);
            } else {
                log.info("保存新的 DevInfoMsg0 数据, processId: {}, data: {}", processId, devInfoMsg0.toString());
                mqttReceiveCallback.devInfoMsg0Service.save(devInfoMsg0);
            }

            log.info("保存 DevInfoMsg0History 数据, processId: {}, data: {}", processId, devInfoMsg0History.toString());
            mqttReceiveCallback.devInfoMsg0HistoryService.save(devInfoMsg0History);

            if (equipments != null && !equipments.isEmpty()) {
                Equipment equipment = equipments.get(0);
                log.info("处理设备告警和故障, processId: {}, 设备SN: {}", processId, equipment.getSn());
                alarmParser.createAlarm(equipment, devInfoMsg0.getW1(), devInfoMsg0.getW2(),
                        mqttReceiveCallback.sptDevLogService, mqttReceiveCallback.workOrdersService,
                        mqttReceiveCallback.redisUtil, mqttReceiveCallback.configInfoService);
                alarmParser.createFault(equipment, devInfoMsg0.getF1(),
                        mqttReceiveCallback.sptDevLogService, mqttReceiveCallback.workOrdersService,
                        mqttReceiveCallback.redisUtil, mqttReceiveCallback.configInfoService);
            }

            log.info("DevInfoMsg0 异步处理完成, processId: {}, SN: {}", processId, devInfoMsg0.getSn());
        } catch (Exception e) {
            log.error("DevInfoMsg0 异步处理发生异常, processId: {}, SN: {}, error: {}", processId, devInfoMsg0.getSn(), e.getMessage(), e);
        }
    }

    /**
     * 异步处理 DevInfoMsg1 数据保存
     */
    @Async("mqttAsyncExecutor")
    public void saveDevInfoMsg1Async(String processId, DevInfoMsg1 devInfoMsg1, DevInfoMsg1 devInfoMsg01,
                                     LambdaQueryWrapper<DevInfoMsg1> devInfoMsg1LambdaQueryWrapper,
                                     DevInfoMsg1History devInfoMsg1History) {
        try {
            log.info("开始异步处理 DevInfoMsg1 数据保存, processId: {}, SN: {}", processId, devInfoMsg1.getSn());

            if (devInfoMsg01 != null) {
                log.info("更新 DevInfoMsg1 数据, processId: {}, data: {}", processId, devInfoMsg1.toString());
                mqttReceiveCallback.devInfoMsg1Service.update(devInfoMsg1, devInfoMsg1LambdaQueryWrapper);
            } else {
                log.info("保存新的 DevInfoMsg1 数据, processId: {}, data: {}", processId, devInfoMsg1.toString());
                mqttReceiveCallback.devInfoMsg1Service.save(devInfoMsg1);
            }

            log.info("保存 DevInfoMsg1History 数据, processId: {}, data: {}", processId, devInfoMsg1History.toString());
            mqttReceiveCallback.devInfoMsg1HistoryService.save(devInfoMsg1History);

            log.info("DevInfoMsg1 异步处理完成, processId: {}, SN: {}", processId, devInfoMsg1.getSn());
        } catch (Exception e) {
            log.error("DevInfoMsg1 异步处理发生异常, processId: {}, SN: {}, error: {}", processId, devInfoMsg1.getSn(), e.getMessage(), e);
        }
    }

    /**
     * 异步处理 DevInfoMsg2 数据保存
     */
    @Async("mqttAsyncExecutor")
    public void saveDevInfoMsg2Async(String processId, DevInfoMsg2 devInfoMsg2, DevInfoMsg2 devInfoMsg02,
                                     LambdaQueryWrapper<DevInfoMsg2> devInfoMsg2LambdaQueryWrapper,
                                     DevInfoMsg2History devInfoMsg2History) {
        try {
            log.info("开始异步处理 DevInfoMsg2 数据保存, processId: {}, SN: {}", processId, devInfoMsg2.getSn());

            if (devInfoMsg02 != null) {
                log.info("更新 DevInfoMsg2 数据, processId: {}, data: {}", processId, devInfoMsg2.toString());
                mqttReceiveCallback.devInfoMsg2Service.update(devInfoMsg2, devInfoMsg2LambdaQueryWrapper);
            } else {
                log.info("保存新的 DevInfoMsg2 数据, processId: {}, data: {}", processId, devInfoMsg2.toString());
                mqttReceiveCallback.devInfoMsg2Service.save(devInfoMsg2);
            }

            log.info("保存 DevInfoMsg2History 数据, processId: {}, data: {}", processId, devInfoMsg2History.toString());
            mqttReceiveCallback.devInfoMsg2HistoryService.save(devInfoMsg2History);

            log.info("DevInfoMsg2 异步处理完成, processId: {}, SN: {}", processId, devInfoMsg2.getSn());
        } catch (Exception e) {
            log.error("DevInfoMsg2 异步处理发生异常, processId: {}, SN: {}, error: {}", processId, devInfoMsg2.getSn(), e.getMessage(), e);
        }
    }

    /**
     * 异步处理 DevInfoMsg3 数据保存
     */
    @Async("mqttAsyncExecutor")
    public void saveDevInfoMsg3Async(String processId, DevInfoMsg3 devInfoMsg3, DevInfoMsg3 devInfoMsg03,
                                     LambdaQueryWrapper<DevInfoMsg3> devInfoMsg3LambdaQueryWrapper,
                                     DevInfoMsg3History devInfoMsg3History) {
        try {
            log.info("开始异步处理 DevInfoMsg3 数据保存, processId: {}, SN: {}", processId, devInfoMsg3.getSn());

            if (devInfoMsg03 != null) {
                log.info("更新 DevInfoMsg3 数据, processId: {}, data: {}", processId, devInfoMsg3.toString());
                mqttReceiveCallback.devInfoMsg3Service.update(devInfoMsg3, devInfoMsg3LambdaQueryWrapper);
            } else {
                log.info("保存新的 DevInfoMsg3 数据, processId: {}, data: {}", processId, devInfoMsg3.toString());
                mqttReceiveCallback.devInfoMsg3Service.save(devInfoMsg3);
            }

            log.info("保存 DevInfoMsg3History 数据, processId: {}, data: {}", processId, devInfoMsg3History.toString());
            mqttReceiveCallback.devInfoMsg3HistoryService.save(devInfoMsg3History);

            log.info("DevInfoMsg3 异步处理完成, processId: {}, SN: {}", processId, devInfoMsg3.getSn());
        } catch (Exception e) {
            log.error("DevInfoMsg3 异步处理发生异常, processId: {}, SN: {}, error: {}", processId, devInfoMsg3.getSn(), e.getMessage(), e);
        }
    }

    /**
     * 异步处理 DevInfoMsg4 数据保存
     */
    @Async("mqttAsyncExecutor")
    public void saveDevInfoMsg4Async(String processId, DevInfoMsg4 devInfoMsg4, DevInfoMsg4 devInfoMsg04,
                                     LambdaQueryWrapper<DevInfoMsg4> devInfoMsg4LambdaQueryWrapper,
                                     DevInfoMsg4History devInfoMsg4History) {
        try {
            log.info("开始异步处理 DevInfoMsg4 数据保存, processId: {}, SN: {}", processId, devInfoMsg4.getSn());

            if (devInfoMsg04 != null) {
                log.info("更新 DevInfoMsg4 数据, processId: {}, data: {}", processId, devInfoMsg4.toString());
                mqttReceiveCallback.devInfoMsg4Service.update(devInfoMsg4, devInfoMsg4LambdaQueryWrapper);
            } else {
                log.info("保存新的 DevInfoMsg4 数据, processId: {}, data: {}", processId, devInfoMsg4.toString());
                mqttReceiveCallback.devInfoMsg4Service.save(devInfoMsg4);
            }

            log.info("保存 DevInfoMsg4History 数据, processId: {}, data: {}", processId, devInfoMsg4History.toString());
            mqttReceiveCallback.devInfoMsg4HistoryService.save(devInfoMsg4History);

            log.info("DevInfoMsg4 异步处理完成, processId: {}, SN: {}", processId, devInfoMsg4.getSn());
        } catch (Exception e) {
            log.error("DevInfoMsg4 异步处理发生异常, processId: {}, SN: {}, error: {}", processId, devInfoMsg4.getSn(), e.getMessage(), e);
        }
    }

    /**
     * 异步处理 DevInfoMsg5 数据保存
     */
    @Async("mqttAsyncExecutor")
    public void saveDevInfoMsg5Async(String processId, DevInfoMsg5 devInfoMsg5, DevInfoMsg5 devInfoMsg05,
                                     LambdaQueryWrapper<DevInfoMsg5> devInfoMsg5LambdaQueryWrapper,
                                     DevInfoMsg5History devInfoMsg5History) {
        try {
            log.info("开始异步处理 DevInfoMsg5 数据保存, processId: {}, SN: {}", processId, devInfoMsg5.getSn());

            if (devInfoMsg05 != null) {
                log.info("更新 DevInfoMsg5 数据, processId: {}, data: {}", processId, devInfoMsg5.toString());
                String bbws = devInfoMsg5.getBbws();
                BBWS bbws1 = new BBWS(bbws);
                bbws1.createAlarm(devInfoMsg5.getSn(), "", bbws1,
                        mqttReceiveCallback.sptDevLogService, mqttReceiveCallback.workOrdersService,
                        mqttReceiveCallback.redisUtil);
                mqttReceiveCallback.devInfoMsg5Service.update(devInfoMsg5, devInfoMsg5LambdaQueryWrapper);
            } else {
                log.info("保存新的 DevInfoMsg5 数据, processId: {}, data: {}", processId, devInfoMsg5.toString());
                mqttReceiveCallback.devInfoMsg5Service.save(devInfoMsg5);
            }

            log.info("保存 DevInfoMsg5History 数据, processId: {}, data: {}", processId, devInfoMsg5History.toString());
            mqttReceiveCallback.devInfoMsg5HistoryService.save(devInfoMsg5History);

            log.info("DevInfoMsg5 异步处理完成, processId: {}, SN: {}", processId, devInfoMsg5.getSn());
        } catch (Exception e) {
            log.error("DevInfoMsg5 异步处理发生异常, processId: {}, SN: {}, error: {}", processId, devInfoMsg5.getSn(), e.getMessage(), e);
        }
    }

    /**
     * 异步处理 DevInfoMsg6 数据保存
     */
    @Async("mqttAsyncExecutor")
    public void saveDevInfoMsg6Async(String processId, DevInfoMsg6 devInfoMsg6, DevInfoMsg6 devInfoMsg06,
                                     LambdaQueryWrapper<DevInfoMsg6> devInfoMsg6LambdaQueryWrapper) {
        try {
            log.info("开始异步处理 DevInfoMsg6 数据保存, processId: {}, SN: {}", processId, devInfoMsg6.getSn());

            if (devInfoMsg06 != null) {
                log.info("更新 DevInfoMsg6 数据, processId: {}, data: {}", processId, devInfoMsg6.toString());
                mqttReceiveCallback.devInfoMsg6Service.update(devInfoMsg6, devInfoMsg6LambdaQueryWrapper);
            } else {
                log.info("保存新的 DevInfoMsg6 数据, processId: {}, data: {}", processId, devInfoMsg6.toString());
                mqttReceiveCallback.devInfoMsg6Service.save(devInfoMsg6);
            }

            log.info("DevInfoMsg6 异步处理完成, processId: {}, SN: {}", processId, devInfoMsg6.getSn());
        } catch (Exception e) {
            log.error("DevInfoMsg6 异步处理发生异常, processId: {}, SN: {}, error: {}", processId, devInfoMsg6.getSn(), e.getMessage(), e);
        }
    }

    /**
     * 用于在设备首次收到消息时，将sn加入redis ListKey
     */
//    public void addSnToRedisList(RedisCache redisUtil, String sn) {
//        if (ObjectUtil.isEmpty(sn)) return;
//        List<String> snList = redisUtil.getCacheList(REDIS_SN_LIST_KEY);
//        if (snList == null) {
//            snList = new ArrayList<>();
//        }
//        if (!snList.contains(sn)) {
//            snList.add(sn);
//            redisUtil.setCacheList(REDIS_SN_LIST_KEY, snList);
//            log.info("设备号 {} 已加入redis ListKey {}", sn, REDIS_SN_LIST_KEY);
//        }
//    }
}

